package shared.runtime

import shared.message._

import org.jgroups.Address

trait ChannelAdaptor extends ResponseHandler {
  import java.util.concurrent._

  private val cache = new ConcurrentHashMap[Long, Synchronizer]
  
  @volatile var totalClients = 0
  
  private trait Synchronizer{
    def handleResponse(r: ResponseMessage)
  }

  private class WaitOneSyncher extends Synchronizer {
    @volatile var holder: Option[ResponseMessage] = None
    def handleResponse(r: ResponseMessage) = {
      this.synchronized {
        holder = Some(r)
        this.notifyAll
      }
    }
  }
  
  private class WaitManySyncher(val max: Int) extends Synchronizer {
    @volatile var list: List[ResponseMessage] = Nil
    
    def handleResponse(r: ResponseMessage) = {
      list = r :: list
      if(list.size == max) {
        this.synchronized {
          this.notifyAll
        }
      }
    }
  }
  
  def send(add: Address, m: Message): Unit
  
  def sendWait[R <: ResponseMessage](add: Address, m: IdentifiedMessage[R]) (timeout: Long): Option[R] = {
    val waiter = new WaitOneSyncher
    cache.put(m.id, waiter)
    waiter.synchronized {
      send(add, m)
      waiter.wait(timeout)
      cache.remove(m.id)
    }
    
    waiter.holder.asInstanceOf[Option[R]]
  }
  
  def scatter(m: Message) = send(null, m)
  
  def scatterGather[R <: ResponseMessage](m: IdentifiedMessage[R]) (timeout: Long): List[R] = {
    val waiter = new WaitManySyncher(totalClients)
    cache.put(m.id, waiter)
    waiter.synchronized{
      scatter(m)
      if(waiter.max > 0) {
        waiter.wait(timeout)
      }
      cache.remove(m.id)
    }
    waiter.list.asInstanceOf[List[R]]
  }

  def handleResponse(r: ResponseMessage) = {
    val waiter = cache.get(r.responseTo)
    waiter.handleResponse(r)
  }
  
  def localAddress: Address
  
  def close = {}
  
}
